package com.arica.mq.consumer.processor;

import com.alibaba.fastjson.JSON;
import com.arica.framework.common.result.ErrorEnum;
import com.arica.framework.common.result.Result;
import com.arica.framework.common.utils.MD5Util;
import com.arica.framework.dto.MqConfigDTO;
import com.arica.framework.dto.MqConsumptionDTO;
import com.arica.project.mq.service.MqConsumptionService;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.yaml.snakeyaml.constructor.DuplicateKeyException;

import java.util.List;

@Service
public class ProcessHelper {
    private static final Logger logger = LoggerFactory.getLogger(ProcessHelper.class);

    @Autowired
    private MqConsumptionService mqConsumptionService;

    /**
     * 判断消息是否幂等
     ***/
    public Result<MqConsumptionDTO> isIdempotent(Object data, Message message, MessageProcessor processor) {
        String processType = processor.getProcessType();
        String businessKey = processor.getBusinessKey(data, message);
        MqConsumptionDTO mqConsumptionDTO = this.buildMqConsumptionDto(processType, businessKey);
        try {
            int count = mqConsumptionService.insert(mqConsumptionDTO);
            if (count == 1) {
                return Result.success(mqConsumptionDTO);
            }
            return Result.fail(ErrorEnum.PARAMS_ERROR.getValue(), "记录插入失败");
        } catch (DuplicateKeyException e1) {
            logger.warn("catch exception isIdempotent 主键冲突，该消息已经处理 dto={}", JSON.toJSONString(mqConsumptionDTO));
            return Result.fail(ErrorEnum.DB_ERROR.getValue(), "主键冲突");
        } catch (Exception e2) {
            logger.error("catch exception 插入数据库异常 dto={}", JSON.toJSONString(mqConsumptionDTO), e2);
            throw new RuntimeException(e2);
        }
    }

    /**
     * 转成业务对应的dto
     *
     * @param messageBody
     * @param sourceClassName
     * @return
     * @throws ClassNotFoundException
     */
    public Object getBusinessData(String messageBody, String sourceClassName) throws ClassNotFoundException {
        return com.alibaba.fastjson.JSON.parseObject(messageBody, Class.forName(sourceClassName));
    }

    /**
     * @param processor
     * @param topic
     * @param tag
     * @return
     */
    public String getSourceClassName(MessageProcessor processor, String topic, String tag) {
        List<MqConfigDTO> mqConfigDTOS = processor.getMqConfig();
        if (CollectionUtils.isEmpty(mqConfigDTOS)) {
            return null;
        }

        String className = "";
        for (MqConfigDTO configDTO : mqConfigDTOS) {
            if (StringUtils.equals(configDTO.getTag(), tag) && StringUtils.equals(configDTO.getTopic(), topic)) {
                className = configDTO.getClassName();
            }
        }
        return className;
    }

    /**
     * 创建
     *
     * @param processType
     * @param businessKey
     * @return
     */
    private MqConsumptionDTO buildMqConsumptionDto(String processType, String businessKey) {
        String md5SourceKey = processType + businessKey;
        String key = MD5Util.getMd5(md5SourceKey);
        return new MqConsumptionDTO(key, processType, businessKey);
    }

}
